package com.ruoyi;

import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.LoggerContext;
import org.apache.kafka.clients.producer.*;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Properties;

/**
 * 生产者
 */
public class ProducerQuickStart {
    static {
        LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
        List<Logger> loggerList = loggerContext.getLoggerList();
        loggerList.forEach(logger -> {
            logger.setLevel(Level.WARN);
        });
    }

    public static void main(String[] args) throws Exception{
        //1.kafka的配置信息
        Properties properties = new Properties();
        //kafka的连接地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.0.80:9092");
        //发送失败，失败的重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG,5);
        //消息key的序列化器
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        //消息value的序列化器
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        //ack确认 0:不确认，速度最快; 1:leader确认;  all: leader和所有follower全部确认
        properties.put(ProducerConfig.ACKS_CONFIG,"all");

        //重试次数，10次没有成功后确认失败
        properties.put(ProducerConfig.RETRIES_CONFIG,10);

        //数据压缩
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");
        //2.生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);

        //封装发送的消息
        ProducerRecord<String,String> record = new ProducerRecord<String, String>("itheima-topic","100001","hello kafka");

        //3.发送消息
        //同步发送，可以获得偏移量，会阻塞
        //RecordMetadata recordMetadata = producer.send(record).get();
        //System.out.println(recordMetadata.offset());

        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {

                if(e!=null){
                    System.out.println("记录异常信息保存到日志表");
                }else{
                    System.out.println(recordMetadata.offset());
                }
            }
        });
        //4.关闭消息通道，必须关闭，否则消息发送不成功
        producer.close();
    }

}
